package com.zyb.lmbackend.controller;

import com.zyb.lmbackend.dto.RunResponseDto;
import com.zyb.lmbackend.dto.TaskStatusDto;
import com.zyb.lmbackend.entity.ClassifyTask;
import com.zyb.lmbackend.service.BatchClassifierService;
import com.zyb.lmbackend.service.ClassifyTaskService;
import io.swagger.v3.oas.annotations.Operation;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
/**
 * 分类任务控制器。
 *
 * 提供：
 * - 启动批量分类任务
 * - 查询任务状态（JSON 或 SSE 实时进度）
 */
@RestController
@RequestMapping("/api/classify")
public class ClassifyController {

    private final BatchClassifierService batchClassifierService;
    private final ClassifyTaskService classifyTaskService;

    public ClassifyController(BatchClassifierService batchClassifierService, ClassifyTaskService classifyTaskService) {
        this.batchClassifierService = batchClassifierService;
        this.classifyTaskService = classifyTaskService;
    }

    @Operation(summary = "启动批量分类任务")
    @PostMapping("/run")
    /**
     * 启动批量分类任务。
     *
     * @param batchSize 每批处理数量（分页抓取未分类记录）
     * @param maxConcurrency 并发执行线程数
     * @param taskVersion 任务版本（为空则创建新版本）
     */
    public ResponseEntity<RunResponseDto> run(
            @RequestParam(required = false) Integer batchSize,
            @RequestParam(required = false) Integer maxConcurrency,
            @RequestParam(required = false) Integer taskVersion) {
//        当第一个大模型生成了分类总结后，这里用线程池异步开启批量读取用户退货的原始数据，再调第二个大模型去匹配第一个大模型的归纳的分类
        ClassifyTask task = batchClassifierService.start(batchSize, maxConcurrency, taskVersion);
        return ResponseEntity.ok(new RunResponseDto(task.getId(), task.getVersion(), task.getStatus()));
    }

    @Operation(summary = "获取任务状态（JSON）。")
    @GetMapping(value = "/task/{id}", produces = MediaType.APPLICATION_JSON_VALUE)
    /**
     * 获取任务状态（JSON）。
     */
    public ResponseEntity<TaskStatusDto> getTaskJson(@PathVariable Long id) {
        ClassifyTask t = classifyTaskService.getById(id);
        if (t == null) {
            return ResponseEntity.notFound().build();
        }
        TaskStatusDto dto = new TaskStatusDto();
        dto.setId(t.getId());
        dto.setVersion(t.getVersion());
        dto.setStatus(t.getStatus());
        dto.setTotal(t.getTotal());
        dto.setProcessed(t.getProcessed());
        dto.setSuccess(t.getSuccess());
        dto.setFailed(t.getFailed());
        dto.setStartedAt(t.getStartedAt());
        dto.setFinishedAt(t.getFinishedAt());
        return ResponseEntity.ok(dto);
    }

    @Operation(summary = "获取任务状态（SSE 实时推送）")
    @GetMapping(value = "/task/{id}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    /**
     * 获取任务状态（SSE 实时推送）。
     *
     * 事件名：
     * - progress：定期推送进度（含 ETA/吞吐）与 15s 心跳
     * - done：任务终止态（SUCCEEDED/FAILED/CANCELLED）
     * - error：异常信息
     */
    public SseEmitter getTaskSse(@PathVariable Long id) {
        SseEmitter emitter = new SseEmitter(0L);
        new Thread(() -> {
            try {
                long lastSentAt = 0L;
                Integer lastProcessed = null;
                while (true) {
                    ClassifyTask t = classifyTaskService.getById(id);
                    if (t == null) {
                        emitter.send(SseEmitter.event().name("error").data("not found"));
                        emitter.complete();
                        return;
                    }
                    int total = t.getTotal() == null ? 0 : t.getTotal();
                    int processed = t.getProcessed() == null ? 0 : t.getProcessed();
                    int success = t.getSuccess() == null ? 0 : t.getSuccess();
                    int failed = t.getFailed() == null ? 0 : t.getFailed();
                    double throughput = 0.0;
                    long etaSeconds = 0L;
                    if (t.getStartedAt() != null) {
                        // 估算吞吐与剩余时间
                        long elapsedSec = java.time.Duration.between(t.getStartedAt(), java.time.LocalDateTime.now()).getSeconds();
                        if (elapsedSec > 0) {
                            throughput = processed / (double) elapsedSec;
                            if (throughput > 0.0 && total > processed) {
                                etaSeconds = Math.max(0L, (long) Math.ceil((total - processed) / throughput));
                            }
                        }
                    }
                    boolean statusFinal = "SUCCEEDED".equalsIgnoreCase(t.getStatus()) ||
                            "FAILED".equalsIgnoreCase(t.getStatus()) ||
                            "CANCELLED".equalsIgnoreCase(t.getStatus());

                    long now = System.currentTimeMillis();
                    boolean heartbeatDue = now - lastSentAt >= 15000; // 15s heartbeat
                    boolean changed = lastProcessed == null || !lastProcessed.equals(processed);
//    还没有完成，并且心跳时间超了15秒，重新单向向前端发送任务完成进度
                    if (changed || heartbeatDue) {
                        // 推送进度事件（前端按 progress 事件名订阅）
//                        后端把任务表现在这里统计后，和前端阅读好的格式转成字符串，前端那边订阅了事件名，这里用sse形式发送，就做到了每隔15秒单向前端发送一下当前的任务完成进度
                        String payload = String.format("{\\\"processed\\\":%d,\\\"success\\\":%d,\\\"failed\\\":%d,\\\"total\\\":%d,\\\"throughput\\\":%.2f,\\\"etaSeconds\\\":%d}",
                                processed, success, failed, total, throughput, etaSeconds);
                        emitter.send(SseEmitter.event().name("progress").data(payload));
                        lastSentAt = now;
                        lastProcessed = processed;
                    }

                    if (statusFinal) {
                        // 结束事件
                        emitter.send(SseEmitter.event().name("done").data(t.getStatus()));
                        emitter.complete();
                        return;
                    }

                    Thread.sleep(1000);
                }
            } catch (Exception e) {
                try { emitter.send(SseEmitter.event().name("error").data(e.getMessage())); } catch (Exception ignore) {}
                emitter.completeWithError(e);
            }
        }, "sse-task-" + id).start();
        return emitter;
    }
}

